package com.atguigu.sql;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gmd
 * @desc 表操作示例。将流数据转换为表，进行SQL操作，然后再转换回流数据。
 * @since 2024-12-04 15:15:18
 */
public class TableStream {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 2L, 2),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3),
                new WaterSensor("s3", 4L, 4)
        );

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 流转表
        Table sensorTable = tableEnv.fromDataStream(sensorDS);
        // 创建视图
        tableEnv.createTemporaryView("sensor", sensorTable);

        // 2. 表转流
        // 2.1 追加流
        Table filterTable = tableEnv.sqlQuery("select id, ts, vc from sensor where ts > 2");
        tableEnv.toDataStream(filterTable, WaterSensor.class).print("filter：");

        // 2.2 changelog流(结果需要更新)
        Table sumTable = tableEnv.sqlQuery("select id, sum(vc) from sensor group by id");
        tableEnv.toChangelogStream(sumTable ).print("sum：");

        // 只要代码中调用了 DataStreamAPI，就需要 execute，否则不需要
        env.execute();
    }

}
